package cn.admin.configcenter.registry.service.client;

import cn.admin.config.common.protocol.Swizzle;
import cn.admin.configcenter.registry.common.config.ConfigSetting;
import cn.admin.configcenter.registry.common.event.Event;
import cn.admin.configcenter.registry.common.event.EventDispatcher;
import cn.admin.configcenter.registry.common.event.EventListener;
import cn.admin.configcenter.registry.common.event.Events;
import cn.admin.configcenter.registry.common.info.ClientConnection;
import cn.admin.configcenter.registry.common.info.GroupId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingDeque;

public class NonPersistDataMerge extends EventListener implements Runnable {

	private static final Logger LOG = LoggerFactory.getLogger(NonPersistDataMerge.class);
	private final BlockingQueue<Event> dataEventQueue = new LinkedBlockingDeque<>();
	private final ConcurrentHashMap<GroupId, ConcurrentLinkedQueue<ClientConnection>> group2conns
			= new ConcurrentHashMap<>();

	public NonPersistDataMerge() {
		Thread t = new Thread(this);
		t.setDaemon(true);
		t.setName(NonPersistDataMerge.class.getName());
		t.start();
	}

	@Override
	public List<Class<? extends Event>> interest() {
		List<Class<? extends Event>> types = new ArrayList<>();
		types.add(Events.NonPersistDataPublishedEvent.class);
		types.add(Events.NonPersistPublisherRemovedEvent.class);
		types.add(Events.ClientDisconnectedEvent.class);
		types.add(Events.GroupDataMannualChangedEvent.class);
		return types;
	}

	@Override
	public void onEvent(Event event) {
		this.dataEventQueue.offer(event);
	}

	public int pendingCount() {
		return this.dataEventQueue.size();
	}

	@Override
	public void run() {
		while (true) {
			try {
				Event event = dataEventQueue.take();
				if (event instanceof Events.NonPersistDataPublishedEvent) {
					Events.NonPersistDataPublishedEvent evt =
							(Events.NonPersistDataPublishedEvent) event;
					processPublish(evt.client, evt.group);
				} else if (event instanceof Events.NonPersistPublisherRemovedEvent) {
					Events.NonPersistPublisherRemovedEvent evt =
							(Events.NonPersistPublisherRemovedEvent) event;
					processUnPublish(evt.client, evt.group);
				} else if (event instanceof Events.ClientDisconnectedEvent) {
					Events.ClientDisconnectedEvent evt =
							(Events.ClientDisconnectedEvent) event;
					processRemoveClient(evt.client);
				} else if (event instanceof Events.GroupDataMannualChangedEvent) {
					Events.GroupDataMannualChangedEvent et =
							(Events.GroupDataMannualChangedEvent) event;
					modifyAndPushDatumData(et.group);
				}
			} catch (InterruptedException e) {
				LOG.warn("[data-merge-error] " + e.toString(), e);
			}
		}
	}

	private void processPublish(ClientConnection client, GroupId groupId) {
		Queue<ClientConnection> contributors = groupContributors(groupId);
		if (!contributors.contains(client)) {
			contributors.add(client);
		}

		EventDispatcher.fireEvent(new Events.GroupDataChangedEvent(groupId));
	}

	private void processUnPublish(ClientConnection client, GroupId group) {
		Queue<ClientConnection> contributors = groupContributors(group);
		contributors.remove(client);
		EventDispatcher.fireEvent(new Events.GroupDataChangedEvent(group));
	}

	private void processRemoveClient(ClientConnection client) {
		Iterator<GroupId> groupIdIter = client.publisherGroupsIteratorR();
		while (groupIdIter.hasNext()) {
			processPublish(client, groupIdIter.next());
		}
	}

	List<GroupId> allDataGroup() {
		return new ArrayList<>(this.group2conns.keySet());
	}

	private ConcurrentLinkedQueue<ClientConnection> groupContributors(GroupId group) {

		ConcurrentLinkedQueue<ClientConnection> result = this.group2conns.get(group);
		if (result == null) {
			result = this.group2conns.putIfAbsent(group, new ConcurrentLinkedQueue<>());
			if (result == null) {
				return this.group2conns.get(group);
			}
		}
		return result;
	}

	private void modifyAndPushDatumData(GroupId group) {

		group.setMannualUpdated(true);
		Collection<ClientConnection> contributors = groupContributors(group);
		if (!contributors.isEmpty()) {
			for (ClientConnection client : contributors) {
				client.updateGroupDataFromConfigService(group);
				processPublish(client, group);
			}
		}
	}

	Map<String,Swizzle> getDatumData(GroupId group) {
		Map<String,Swizzle> result = new HashMap<>();
		Collection<ClientConnection> contributors = groupContributors(group);
		if (contributors.isEmpty()) {
			return result;
		} else {
			for (ClientConnection client : contributors) {
				result.put(client.originalHostId(), client.groupData(group));
			}
			return result;
		}
	}

	static boolean needFilterInMiniCenter(ClientConnection client,GroupId group) {
		return client.isHSFInfo(group) ||
				client.isNotifyInfo(group) ||
				ConfigSetting.citySeparateSet.contains(group.dataId);
	}

}
